package git.soulbgm.controller;

import git.soulbgm.utils.Counter;
import git.soulbgm.utils.TimeInterval;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;

import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 生产控制器
 *
 * @author SoulBGM
 * @date 2022/01/13
 */
@RestController
public class ProducerController {

    @Autowired
    private KafkaTemplate<String, String> kafka;

    @Autowired
    private RestTemplate restTemplate;

    @Value("${config.nodes}")
    private List<String> nodes;

    @Value("${config.testData}")
    private String testData;

    private final String KAFKA_TOPIC = "kafka-test";
    private String http = "http://";
    ;

    @GetMapping("send")
    public String send(int num, int threadNum) throws InterruptedException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        TimeInterval interval = TimeInterval.get();
        long startTime = interval.start();
        int count = num * threadNum;
        CountDownLatch latch = new CountDownLatch(count);
        ExecutorService pool = Executors.newFixedThreadPool(threadNum);
        for (int i = 0; i < threadNum; i++) {
            pool.submit(() -> {
                for (long j = 0; j < num; j++) {
                    String data = Thread.currentThread().getName() + "$&" + testData + "$&" + j;
                    ListenableFuture<SendResult<String, String>> future = kafka.send(KAFKA_TOPIC, data);
                    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                        @Override
                        public void onFailure(Throwable throwable) {
                            latch.countDown();
                        }

                        @Override
                        public void onSuccess(SendResult<String, String> stringStringSendResult) {
                            latch.countDown();
                        }
                    });
                }
            });
        }
        latch.await();
        long diff = interval.interval();
        String startTimeStr = sdf.format(new Date(startTime));
        String endTimeStr = sdf.format(new Date(startTime + diff));
        return "发送成功" + count + "条消息, 开始执行时间:" + startTimeStr + ", 结束执行时间:" + endTimeStr + ", 总耗时" + diff + "毫秒";
    }

    @GetMapping("count")
    public String count() {
        return "已计数:" + Counter.count();
    }

    @GetMapping("lastTime")
    public String lastTime() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        return "开始计数时间:" + sdf.format(new Date(Counter.startTime())) + " 最后计数时间:" + sdf.format(new Date(Counter.lastTime()));
    }

    @GetMapping("recount")
    public String recount() {
        Counter.makeZero();
        return "已将计算器归零";
    }

    @GetMapping("countAll")
    public String countAll() {
        return getResult("/count");
    }

    @GetMapping("lastTimeAll")
    public String lastTimeAll() {
        return getResult("/lastTime");
    }

    @GetMapping("recountAll")
    public String recountAll() {
        return getResult("/recount");
    }

    /**
     * 得到远程的结果
     *
     * @param url url
     * @return {@link String}
     */
    private String getResult(String url) {
        StringBuilder sb = new StringBuilder();
        for (String node : nodes) {
            String result = restTemplate.getForObject(http + node + url, String.class);
            sb.append(node);
            sb.append(" ");
            sb.append(result);
            sb.append("\n");
        }
        return sb.toString();
    }

}
